package org.jsbd.boss.service.impl;

import org.jsbd.boss.common.redis.RedisClient;
import org.jsbd.boss.service.ITaskPoolService;
import org.jsbd.boss.util.CacheKey;
import org.jsbd.boss.util.SpringContextUtils;
import org.jsbd.boss.util.threads.TaskPoolFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;

/**
 * @desc
 * @Date 2018/9/10 15:06
 * @Author xupengcheng [xupengcheng@zhexinit.com]
 */
@Service
public class TaskPoolServiceImpl implements ITaskPoolService {

    public static final Logger logger = LoggerFactory.getLogger(TaskPoolServiceImpl.class);

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(30, 100, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(500), new TaskPoolFactory("taskpool"), new RejectedHandler(CacheKey.A_REJECT_TASK_QUEUE));

    private static ThreadPoolExecutor retentTaskUpLoadExecutor = new ThreadPoolExecutor(30, 100, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(500), new TaskPoolFactory("retentTaskUpLoad"), new RejectedHandler(CacheKey.AR_REJECT_TASK_QUEUE));

    private static ThreadPoolExecutor retentTaskPoolExecutor = new ThreadPoolExecutor(30, 100, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), new TaskPoolFactory("retentTaskPool"), new ThreadPoolExecutor.AbortPolicy());

    private static ThreadPoolExecutor retentTaskQueueExecutor = new ThreadPoolExecutor(30, 100, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), new TaskPoolFactory("retentTaskQueue"), new ThreadPoolExecutor.AbortPolicy());


    private static ThreadPoolExecutor taskTimerExecutor = new ThreadPoolExecutor(8, 30, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(30), new TaskPoolFactory("tasktimer"), new ThreadPoolExecutor.AbortPolicy());


    @Override
    public void exec(Runnable task) {
        threadPoolExecutor.execute(task);
    }

    @Override
    public void execRetentPoolTask(Runnable task) {
        retentTaskPoolExecutor.execute(task);
    }

    @Override
    public void execRetentQueueTask(Runnable task) {
        retentTaskQueueExecutor.execute(task);
    }

    @Override
    public void execRetentUploadTask(Runnable task) {
        retentTaskUpLoadExecutor.execute(task);
    }

    @Override
    public void execTaskTimer(Runnable task) {
        taskTimerExecutor.execute(task);
    }


    public void stop() {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                if (threadPoolExecutor != null && !threadPoolExecutor.isShutdown()) {
                    threadPoolExecutor.shutdown();
                }

                if (retentTaskPoolExecutor != null && !retentTaskPoolExecutor.isShutdown()) {
                    retentTaskPoolExecutor.shutdown();
                }


                if (taskTimerExecutor != null && !taskTimerExecutor.isShutdown()) {
                    taskTimerExecutor.shutdown();
                }


                if (retentTaskUpLoadExecutor != null && !retentTaskUpLoadExecutor.isShutdown()) {
                    retentTaskUpLoadExecutor.shutdown();
                }

                if (retentTaskQueueExecutor != null && !retentTaskQueueExecutor.isShutdown()) {
                    retentTaskQueueExecutor.shutdown();
                }
            }
        });
    }

    public void execReject() {
        new Thread(new RejectTask(CacheKey.A_REJECT_TASK_QUEUE, threadPoolExecutor)).start();
        new Thread(new RejectTask(CacheKey.AR_REJECT_TASK_QUEUE, retentTaskUpLoadExecutor)).start();
    }


    class RejectTask implements Runnable {

        private String key;
        private ThreadPoolExecutor poolExecutor;

        public RejectTask(String key, ThreadPoolExecutor poolExecutor) {
            this.key = key;
            this.poolExecutor = poolExecutor;
        }

        @Override
        public void run() {
            try {
                //等spring启动完成
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            while (true) {
                try {
                    RedisClient redisClient = SpringContextUtils.getBean("redisClient");
                    Runnable task = (Runnable) redisClient.pollFromQueue(key);
                    if (task == null) {
                        Thread.sleep(1000);
                    } else {
                        poolExecutor.execute(task);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

//    @PostConstruct
    private void initTask() {
        execReject();
        stop();
    }

    private static class RejectedHandler implements RejectedExecutionHandler {

        private String key;

        public RejectedHandler(String key) {
            this.key = key;
        }

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            logger.warn("任务被拒绝 {}", r);
            RedisClient redisClient = SpringContextUtils.getBean("redisClient");
            redisClient.putToQueue(key, r);
        }
    }
}
